package com.jxd.hbase.utils;


import com.jxd.config.ApplicationContextRegister;
import com.jxd.config.HbaseConfig;
import com.jxd.entity.MesContent;
import com.jxd.entity.MesHead;
import com.jxd.hbase.analysis.FaultStatistics;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.context.WebApplicationContext;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Created with IDEA
 * @author:zhaoyuZhang
 * @Date:2020/7/22 10:54
 * @program: bigdata
 */
public class AddHbase {
    @Value("${hbase.zookeeper.property.clientPort}")
    private static String clientPort;
    @Value("${hbase.zookeeper.quorum}")
    private static String clientQuorum;
    private static Configuration configuration;
    private static  Connection connection = null;
    private static Admin admin ;
    private static  String tableName =null;
    private  static BufferedMutator table =null;
    public static  Connection getHbaseConnection(){
        configuration= HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort","2181");
        //引用外部配置问件 现在先行调试 zookeeper 服务器地址
        configuration.set("hbase.zookeeper.quorum","192.168.127.12");
        ExecutorService threads = Executors.newFixedThreadPool(20);//手动？
        try {
            connection = ConnectionFactory.createConnection(configuration, threads);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("链接创建失败，异常信息为"+e);
        }
        return connection;
    }

    /**
     * @author:zhaoyuZhang
     * @date  2020/07/22
     * @param tablename
     * 向Content中插入数据
     */
    public static void addHbaseContent(String tablename, List<MesContent> mesContents){
        WebApplicationContext ctx = (WebApplicationContext) ApplicationContextRegister.getApplicationContext();
        HbaseConfig hbaseConfig =(HbaseConfig)ctx.getBean(HbaseConfig.class);
         connection = hbaseConfig.HbaseConnection();
        System.out.println(mesContents.size());
        try {
            List<Mutation> mutationList = new ArrayList<>();
             table =
                    connection.getBufferedMutator(TableName.valueOf(tablename));
            for (int i = 0; i <mesContents.size() ; i++) {
                Put p = new Put(Bytes.toBytes(mesContents.get(i).getRowkey()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesSignId"),Bytes.toBytes(mesContents.get(i).getMesSignId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("id"),Bytes.toBytes(mesContents.get(i).getId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("code"),Bytes.toBytes(mesContents.get(i).getCode()));
                mutationList.add(p);
            }
            table.mutate(mutationList);

        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            IOUtils.closeStream(table);
            //关闭链接
            IOUtils.closeStream(connection);
        }

    }
    public static void addHbaseHead(String tablename, List<MesHead> mesHeads){
        WebApplicationContext ctx = (WebApplicationContext) ApplicationContextRegister.getApplicationContext();
        HbaseConfig hbaseConfig =(HbaseConfig)ctx.getBean(HbaseConfig.class);
        connection = hbaseConfig.HbaseConnection();
        try {
            List<Mutation> mutationList = new ArrayList<>();
            table =
                    connection.getBufferedMutator(TableName.valueOf(tablename));
            for (int i = 0; i <mesHeads.size() ; i++) {
                Put p = new Put(Bytes.toBytes(mesHeads.get(i).getKeyWord()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesDate"),Bytes.toBytes(mesHeads.get(i).getMesDate()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesTime"),Bytes.toBytes(mesHeads.get(i).getMesTime()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesSignId"),Bytes.toBytes(mesHeads.get(i).getMesSignId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesId"),Bytes.toBytes(mesHeads.get(i).getMesId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesType"),Bytes.toBytes(mesHeads.get(i).getMesType()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesSeq"),Bytes.toBytes(mesHeads.get(i).getMesSignId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("mesSize"),Bytes.toBytes(mesHeads.get(i).getMesSignId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("sendDevice"),Bytes.toBytes(mesHeads.get(i).getSendDevice()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("receivingDevice"),Bytes.toBytes(mesHeads.get(i).getReceivingDevice()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("shipId"),Bytes.toBytes(mesHeads.get(i).getMesSignId()));
                p.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("fileName"),Bytes.toBytes(mesHeads.get(i).getMesSignId()));
                mutationList.add(p);
            }
            table.mutate(mutationList);

        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            IOUtils.closeStream(table);
            //关闭链接
            IOUtils.closeStream(connection);
        }

    }


}
